
//Copyright 1997-2009 Syrinx Development, Inc.
//This file is part of the Syrinx Web Application Framework (SWAF).
// == BEGIN LICENSE ==
//
// Licensed under the terms of any of the following licenses at your
// choice:
//
//  - GNU General Public License Version 3 or later (the "GPL")
//    http://www.gnu.org/licenses/gpl.html
//
//  - GNU Lesser General Public License Version 3 or later (the "LGPL")
//    http://www.gnu.org/licenses/lgpl.html
//
//  - Mozilla Public License Version 1.1 or later (the "MPL")
//    http://www.mozilla.org/MPL/MPL-1.1.html
//
// == END LICENSE ==
using System;
using System.Diagnostics;
using System.Collections;
using System.Threading;
using System.Xml;

using Swaf;
using Swaf.Call;
using Swaf.Container;
using Swaf.BizRule;

namespace Swaf.BizRule.ThreadedQueue
{
	/// <summary>
	/// Summary description for RuleQueue.
	/// </summary>
	public class RuleQueue : IRuleQueue
	{
		private int m_numElInQ = 0;
		protected IRuleMonitor m_monitor;
		/// <summary>The list of executing initial requests that want threads.</summary>
		/// <remarks>The list is always ordered the head of the list containing all the 
		/// initial requests that need threads followed by all the requests that want threads.</remarks>
		protected LinkedList m_waitingInitialRequests;
		protected LinkedList.NodeRef m_wantingPos;
		//MJD5: protected LinkedList m_ruleQueue;
		private Semaphore m_workAvailSem;
		private IList m_workers;
		int m_minThreads = 5;
		int m_maxThreads = 20;
		bool m_stopped = true;
		LocalDataStoreSlot m_workerSlot;
#if(!COMPACT_FRAMEWORK)
		private ApartmentState m_workerApartmentState = ApartmentState.Unknown;
#endif

		public uint ElementsInQueue {get{return (uint)m_workAvailSem.count;}}

		public RuleQueue(XmlElement initInfo)
		{
			if(initInfo.Name == "QueueParams")
			{
				m_minThreads = Int32.Parse(initInfo.GetAttribute("minThreads"));
				m_maxThreads = Int32.Parse(initInfo.GetAttribute("maxThreads"));
				try
				{
					XmlAttribute aptType = initInfo.GetAttributeNode("comApartmentType");
#if(!COMPACT_FRAMEWORK)
					if(aptType != null)
						m_workerApartmentState = (ApartmentState)Enum.Parse(typeof(ApartmentState), aptType.Value, true);
#endif
				}
				catch(Exception)
				{
					//An exception is thrown when the comApartmentType attribute value is not
					//a valid System.Threading.ApartmentState enumeration value.

					//TODO Log exception
				}
				int minWorkersPerRequest = -1;
				XmlAttribute attr = initInfo.GetAttributeNode("minWorkersPerRequest");
				if(attr != null)
					minWorkersPerRequest = Int32.Parse(attr.Value);
				BaseInitialRequest.DefaultMinWorkers = minWorkersPerRequest;

				int maxWorkersPerRequest = -1;
				attr = initInfo.GetAttributeNode("maxWorkersPerRequest");
				if(attr != null)
					maxWorkersPerRequest = Int32.Parse(attr.Value);
				BaseInitialRequest.DefaultMaxWorkers = maxWorkersPerRequest;

				m_workers = new ArrayList();

				m_waitingInitialRequests = new LinkedList();
				m_wantingPos = m_waitingInitialRequests.addHead(null);

				//MJD5: m_ruleQueue = new LinkedList();
				m_workAvailSem = new Semaphore(0);
				m_workerSlot = Thread.GetNamedDataSlot("RuleQueueWorker");
				//m_workerSlot = Thread.AllocateNamedDataSlot("RuleQueueWorker");

				manageWorkers();
				m_stopped = false;
			}
		}

		public IRuleMonitor Monitor
		{
			get
			{
				return m_monitor;
			}
			set
			{
				m_monitor = value;
			}
		}


		internal LocalDataStoreSlot WorkerSlot
		{
			get{return m_workerSlot;}
		}

		public void stop()
		{
			foreach(Worker w in m_workers)
				w.stopProcessing(false);
			m_workAvailSem.releaseAll();
		}

		public uint WorkerCount 
		{
			get
			{
				return (uint)m_workers.Count;
			}
			set
			{
				if(value == m_maxThreads)
					return;
				m_maxThreads = (int)value;
				manageWorkers();
			}
		}

		public IRuleGroup enqueue(IRuleInfo ruleInfo, ICallResults callInfo, IRuleContext context, IRuleGroup group, ICall callback, object ruleParams)
		{
			if(m_stopped)
				throw new Exception("Rule Queue is stopped and cannot be reused.");

			IRuleGroup realGroup = getRuleGroupForNextCall(group, callInfo, context.App, true);
			QueueElement qEl  = new QueueElement(this, ruleInfo, callInfo, context, realGroup, callback, ruleParams);
			//MJD5: checkRulesetWaiting((BaseInitialRequest)realGroup.MainRequest);
			return enqueue(qEl);
		}

		protected IRuleGroup getRuleGroupForNextCall(IRuleGroup group, ICallResults info, IApplication app, bool createGroup)
		{
			Debug.Assert(info is SingleCallAndResult);
			if(group != null)
				return group;

			Worker w = (Worker)Thread.GetData(WorkerSlot);
			ResultsRuleGroup newGroup = null;
			if(w == null)
			{
				newGroup =  new BaseInitialRequest(app);
			}
			else
			{
				if(createGroup)
					newGroup = (ResultsRuleGroup)w.CurrentRuleGroup.Group.createChildGroup(true);
				else
					return w.CurrentRuleGroup.Group;
			}
			newGroup.addResult((SingleCallAndResult)info);
			return newGroup;
		}

		internal void checkRulesetWaiting(BaseInitialRequest ruleset)
		{
			//Debug.Assert(ruleset != null);
			if(ruleset != null && ruleset.calcNumThreadsNeeded(true) > 0 && ruleset.QueueWaitingPos == null)
			{
				lock(m_waitingInitialRequests)
				{
					//This check ensures that 2 threads working on the same ruleset dont both put the ruleset in the waiting list.
					if(ruleset.QueueWaitingPos == null)
					{
						Debug.WriteLine("Adding ruleset to waiting for threads list");
						bool isWanting = ruleset.calcNumThreadsNeeded(true) !=0;
						bool isNeeding = ruleset.calcNumThreadsNeeded(false) !=0;
						if(isNeeding)
							ruleset.QueueWaitingPos = m_waitingInitialRequests.addHead(ruleset);
						else if(isWanting)
							ruleset.QueueWaitingPos = m_waitingInitialRequests.insert(ruleset, m_wantingPos, true);
					}
				}
			}
		}

		protected internal uint calcNumNeededThreads()
		{
			return calcNumNeededThreads(false);
		}

		protected internal uint calcNumNeededThreads(bool useMax)
		{
			uint numNeeded = 0;
			lock(m_waitingInitialRequests)
			{
				LinkedList.NodeRef node = m_waitingInitialRequests.Head;
				while(node != null)
				{
					if(node.Element != null) //null here is possible due to needPos marker node being in list now.
						numNeeded += ((BaseInitialRequest)node.Element).calcNumThreadsNeeded(useMax);
					node = node.Next;
				}
			}
			return numNeeded;
		}

		public IRuleGroup enqueue(IRuleInfo ruleInfo, ICallResults callInfo, IRuleContext context, IRuleGroup group,
									object ruleParams, ref bool executeNowIfWorker)
		{
			if(m_stopped)
				throw new Exception("Rule Queue is stopped and cannot be reused.");

			Worker w = (Worker)Thread.GetData(WorkerSlot);
			//QueueElement qEl = new QueueElement(this, ruleInfo, callInfo, context, getRuleGroupForNextCall(group, callInfo, context.App), ruleParams);
			//checkRulesetWaiting((w==null)?null:w.CurrentRequest);
			bool executeNow = executeNowIfWorker && w != null;
			IRuleGroup realGroup = getRuleGroupForNextCall(group, callInfo, context.App, !executeNow);
			QueueElement qEl  = new QueueElement(this, ruleInfo, callInfo, context, realGroup, null, ruleParams, !executeNow);
			if(executeNow)
			{
				
				if(qEl.RulesetQueuePos != null)
					w.CurrentRequest.pullFromRuleQueue(qEl, null);
				((IRuleGroupControl)qEl.ruleGroup).ruleAdded();
				qEl.execute(w);
				return qEl.ruleGroup;
			}
			else			
			{
				executeNowIfWorker = false;
				return enqueue(qEl);
			}
		}

		private IRuleGroup enqueue(QueueElement el)
		{
			try
			{
				((BaseInitialRequest)el.ruleGroup.MainRequest).addRule(el);
				((IRuleGroupControl)el.ruleGroup).ruleAdded();
				checkRulesetWaiting((BaseInitialRequest)el.ruleGroup.MainRequest);
				//MJD5:
				//m_workAvailSem.release();
				m_workAvailSem.setAbsolute(calcNumNeededThreads(true));

				/*MJD5:
				lock(m_ruleQueue)
				{
					el.MainRuleQueuePos = m_ruleQueue.addHead(el);
					++m_numElInQ;
					((BaseInitialRequest)el.ruleGroup.MainRequest).addRule(el);
					((IRuleGroupControl)el.ruleGroup).ruleAdded();
					checkRulesetWaiting((BaseInitialRequest)el.ruleGroup.MainRequest);
					//MJD5:
					//m_workAvailSem.release();
					m_workAvailSem.setAbsolute(calcNumNeededThreads(true));
				}	
				*/
				return el.ruleGroup;
			}
			catch (Exception)
			{
				throw new BadRuleCallException(el.ruleInfo.CallInfo.Id);
			}
		}

		internal QueueElement dequeue(Worker w, bool waitForWork)
		{
			QueueElement el = null;
			BaseInitialRequest rs = w.CurrentRequest;
			if(rs != null)
				el = rs.dequeue(w, false);
			else
			{
				if(m_workAvailSem.acquire(waitForWork?-1:0) == false)
					return null;
				el = getNextQueueElement(w);
				/*
				if(el == null)
					m_workAvailSem.release();
				*/
			}
			return el;
				
			/*
			if(m_workAvailSem.acquire(waitForWork?-1:0) == false)
				return null;

			QueueElement el = null;

			try
			{
				lock(m_ruleQueue)
				{
					BaseInitialRequest rs = w.CurrentRequest;
					if(rs != null)
						el = rs.dequeue(w, false);	
					if(el == null)
						el = getNextQueueElement(w);
					//Debug.Assert(el != null);
					if(el == null)
					{
						el = null;
					}
					else
						--m_numElInQ;
				}
			
				return el;
			}
			finally
			{
				//If no queue element was returned for some reason then we need to
				//'put back' the count we took from the m_workAvailSem
				if(el == null)
					m_workAvailSem.release();
			}
			*/
		}

		private QueueElement getNextQueueElement(Worker w)
		{
			QueueElement el = null;
			lock(m_waitingInitialRequests)
			{
				if(m_waitingInitialRequests.Head != m_waitingInitialRequests.Tail) //This should only happen where the only thing in the list is the wantingPos marker
				{
					bool wasNeed = true;
					LinkedList.NodeRef nref = m_wantingPos.Prev; //First try and get the next available initial requests that needs threads
					if(nref == null) //No more initial requests that need work.
					{
						nref = m_waitingInitialRequests.Tail;  //Get the next available initial request that wants threads
						wasNeed = false;
					}

					//MJD5: LinkedList.NodeRef nref = m_waitingInitialRequests.Tail;
					if(nref != null)  //This should never be true because of outer most if statement
					{
						BaseInitialRequest rcall = (BaseInitialRequest)nref.Element;
						el = rcall.dequeue(w, true);

						bool needsNow = rcall.calcNumThreadsNeeded(false) != 0;
						bool wantsNow = rcall.calcNumThreadsNeeded(true) != 0;

						if(!wantsNow)
						{
							Debug.WriteLine("Removing ruleset from waiting for threads list");
							m_waitingInitialRequests.remove(nref);
							rcall.QueueWaitingPos = null;
						}
						else
						{
							if(needsNow && !wasNeed)
								m_waitingInitialRequests.moveNode(nref, m_wantingPos, false);
							else if(wasNeed && !needsNow)
								m_waitingInitialRequests.moveNode(nref, m_waitingInitialRequests.Tail, true);
						}
					}
				}
			}
			return el;
		}

		private BaseInitialRequest getNextNeedingRequest()
		{
			BaseInitialRequest req = null;

			return req;
		}

		private BaseInitialRequest getNextWantingRequest()
		{
			BaseInitialRequest req = null;

			return req;
		}

		public IRunningRule CurrentThreadRunningRule
		{
			get
			{
				IRunningRule r = null;

				Worker w = (Worker)Thread.GetData(WorkerSlot);
				if(w != null)
					r = w.CurrentRunningRule;
				
				return r;
			}
		}

		protected void manageWorkers()
		{
			//TODO: For this development cycle, just make the max number of threads.
			//Later this code should change to grow or shink the number of workers in the list 
			//based on queue size, cpu utilization(?), and current threads operating.
			if(m_workers.Count < m_maxThreads)
			{
				int numNewWorkers = m_maxThreads - m_workers.Count;
				for(int p = 0; p < numNewWorkers; ++p)
				{
#if(!COMPACT_FRAMEWORK)
					Worker w = new Worker(this, m_workerApartmentState);
#else
					Worker w = new Worker(this);
#endif
					m_workers.Add(w);
				}
			}
			else
			{
				//Remove workers
				int numToRemove = m_workers.Count - m_maxThreads;
				for(int p = 0; p < numToRemove; ++p)
				{
					Worker w = (Worker)m_workers[m_workers.Count -1];
					m_workers.RemoveAt(m_workers.Count -1);
					w.stopProcessing(false);
				}
			}
		}

		/* MJD5:
		internal void pullFromRuleQueue(QueueElement el)
		{
			lock(m_ruleQueue)
			{
				if(el.MainRuleQueuePos != null)
					m_ruleQueue.remove(el.MainRuleQueuePos);
			}
		}
		*/
	}
}
